package com.base.txl.utils;

import android.app.Activity;
import android.support.annotation.NonNull;
import com.fan.txl.data.base.Rsp;
import java.util.concurrent.TimeUnit;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;
import io.reactivex.internal.functions.Functions;
import io.reactivex.schedulers.Schedulers;
/**
 * Created by jiajiabao on 17-1-10.
 */

public class RxUtils {


    private RxUtils() { }

    public static final ObservableTransformer THREAD_TRANSFORMER = new ObservableTransformer() {
        @Override public ObservableSource apply(Observable upstream) {
            return upstream.subscribeOn(Schedulers.newThread())
                    .unsubscribeOn(Schedulers.newThread())
                    .observeOn(Schedulers.newThread());
        }
    };

    public static final ObservableTransformer THREAD_ON_UI_TRANSFORMER =
            new ObservableTransformer() {
                @Override public ObservableSource apply(Observable upstream) {
                    return upstream.subscribeOn(Schedulers.newThread())
                            .unsubscribeOn(Schedulers.newThread())
                            .observeOn(AndroidSchedulers.mainThread());
                }
            };

    public static final ObservableTransformer IO_TRANSFORMER = new ObservableTransformer() {
        @Override public ObservableSource apply(Observable upstream) {
            return upstream.subscribeOn(Schedulers.io())
                    .unsubscribeOn(Schedulers.io())
                    .observeOn(Schedulers.io());
        }
    };




    public static final ObservableTransformer IO_ON_UI_TRANSFORMER = new ObservableTransformer() {
        @Override public ObservableSource apply(Observable upstream) {
            return upstream.subscribeOn(Schedulers.io())
                    .unsubscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread());
        }
    };

    public static final ObservableTransformer COMPUTATION_TRANSFORMER =
            new ObservableTransformer() {
                @Override public ObservableSource apply(Observable upstream) {
                    return upstream.subscribeOn(Schedulers.computation())
                            .unsubscribeOn(Schedulers.newThread())
                            .observeOn(Schedulers.newThread());
                }
            };

    public static final ObservableTransformer COMPUTATION_ON_UI_TRANSFORMER =
            new ObservableTransformer() {
                @Override public ObservableSource apply(Observable upstream) {
                    return upstream.subscribeOn(Schedulers.computation())
                            .unsubscribeOn(Schedulers.newThread())
                            .observeOn(AndroidSchedulers.mainThread());
                }
            };

    @SuppressWarnings("unchecked")
    public static <T> ObservableTransformer<T, T> applySchedulers(ObservableTransformer transformer) {
        return (ObservableTransformer<T, T>) transformer;
    }

    /**
     * 运行一个定时任务在子线程
     *
     * @param delay 延迟时间，单位：毫秒
     * @param onNext
     * @return
     */
    public static Disposable time(long delay, @NonNull Consumer<Long> onNext) {
        return time(delay, TimeUnit.MILLISECONDS, onNext);
    }

    /**
     * 运行一个定时任务在子线程
     *
     * @param delay 延迟时间
     * @param unit 单位
     * @param onNext
     * @return
     */
    public static Disposable time(long delay, TimeUnit unit, @NonNull Consumer<Long> onNext) {
        return Observable.timer(delay, unit)
                .compose(RxUtils.<Long>applySchedulers(COMPUTATION_TRANSFORMER))
                .subscribe(onNext);
    }

    /**
     * 运行一个定时任务在UI线程
     *
     * @param delay 延迟时间
     * @param unit 单位
     * @param onNext
     * @return
     */
    public static Disposable timeOnUI(
            long delay, TimeUnit unit, @NonNull Consumer<Long> onNext) {
        return Observable.timer(delay, unit)
                .compose(RxUtils.<Long>applySchedulers(COMPUTATION_ON_UI_TRANSFORMER))
                .subscribe(onNext);
    }

    /**
     * 运行一个轮询任务在子线程
     *
     * @param interval 轮询间隔，单位：毫秒
     * @param onNext
     * @return
     */
    public static Disposable interval(long interval, @NonNull Consumer<Long> onNext) {
        return interval(interval, TimeUnit.MILLISECONDS, onNext);
    }

    /**
     * 运行一个轮询任务在子线程
     *
     * @param interval 轮询间隔
     * @param unit 单位
     * @param onNext
     * @return
     */
    public static Disposable interval(
            long interval, TimeUnit unit, @NonNull Consumer<Long> onNext) {
        return Observable.interval(interval, unit)
                .compose(RxUtils.<Long>applySchedulers(COMPUTATION_TRANSFORMER))
                .subscribe(onNext);
    }

    /**
     * 运行一个任务在子线程
     *
     * @param backgroundAction
     * @return
     */
    public static Disposable run(@NonNull Action backgroundAction) {
        return Observable.<Integer>empty()
                .compose(RxUtils.<Integer>applySchedulers(THREAD_TRANSFORMER))
                .subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER,
                        backgroundAction);
    }

    /**
     * 运行一个任务在UI线程
     *
     * @param backgroundSubscribe
     * @param uiAction
     * @return
     */
    public static <T> Disposable runOnUI(
            @NonNull ObservableOnSubscribe<T> backgroundSubscribe,
            @NonNull Consumer<T> uiAction) {
        return Observable.create(backgroundSubscribe)
                .compose(RxUtils.<T>applySchedulers(THREAD_ON_UI_TRANSFORMER))
                .subscribe(uiAction);
    }

    /**
     * @param observable 被观察者
     * @return Observable 被观察者 转换后还可以做其他的处理
     */
    public static Observable observable(@NonNull Observable observable, final Activity mActivity) {
        return  observable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .filter(new Predicate<Rsp>() {

                    @Override
                    public boolean test(Rsp rsp) throws Exception {
                        boolean status = rsp == null ? false : rsp.getCode() == Rsp.OK;
                        if (!status) {
                            // Prompt.getInstance().toast(mActivity, rsp.getMsg());
                        }
                        return status;
                    }
                });
    }

    /**
     * @param observable 被观察者
     * @param observer   观察者
     * @return Subscription 转换后不需要做其他处理
     */
    public static Disposable observable(@NonNull Observable observable, @NonNull Observer observer) {
        //2.0
        //new Consumer<>()
       return observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe();
        //1.x
       /* return observable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(observer)
                ;*/
    }
}
